package com.raft.kvstore.server;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeoutException;

import javax.annotation.Nonnull;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.com.caucho.hessian.io.HessianInput;
import com.alibaba.com.caucho.hessian.io.HessianOutput;
import com.raft.core.log.statemachine.AbstractSingleThreadStateMachine;
import com.raft.core.node.Node;
import com.raft.core.node.role.RoleName;
import com.raft.core.node.role.RoleNameAndLeaderId;
import com.raft.core.node.task.GroupConfigChangeTaskReference;
import com.raft.core.service.AddNodeCommand;
import com.raft.core.service.RemoveNodeCommand;
import com.raft.kvstore.message.CommandRequest;
import com.raft.kvstore.message.Failure;
import com.raft.kvstore.message.GetCommand;
import com.raft.kvstore.message.GetCommandResponse;
import com.raft.kvstore.message.Redirect;
import com.raft.kvstore.message.SetCommand;
import com.raft.kvstore.message.Success;

/**kv服务实现，持有node实现
 * @author zhangsulei
 *
 */
public class Service {

    private static final Logger logger = LoggerFactory.getLogger(Service.class);
    private final Node node;
    // 请求id和连接的映射，用于在核心组件执行完成后回调客户端
    private final ConcurrentMap<String, CommandRequest<?>> pendingCommands = new ConcurrentHashMap<>();
    // kv服务的数据
    private Map<String, byte[]> map = new HashMap<>();

    public Service(Node node) {
        this.node = node;
        // 注册状态机到node实例
        this.node.registerStateMachine(new StateMachineImpl());
    }

    /**添加节点
     * @param commandRequest
     */
    public void addNode(CommandRequest<AddNodeCommand> commandRequest) {
        // 确认是leader节点
    	Redirect redirect = checkLeadership();
        if (redirect != null) {
            commandRequest.reply(redirect);
            return;
        }

        AddNodeCommand command = commandRequest.getCommand();
        GroupConfigChangeTaskReference taskReference = this.node.addNode(command.toNodeEndpoint());
        awaitResult(taskReference, commandRequest);
    }

    /**等待结果
     * @param taskReference
     * @param commandRequest
     */
    private <T> void awaitResult(GroupConfigChangeTaskReference taskReference, CommandRequest<T> commandRequest) {
        try {
        	// 等待3秒
            switch (taskReference.getResult(3000L)) {
                case OK:
                    commandRequest.reply(Success.INSTANCE);
                    break;
                case TIMEOUT:
                    commandRequest.reply(new Failure(101, "timeout"));
                    break;
                default:
                    commandRequest.reply(new Failure(100, "error"));
            }
        } catch (TimeoutException e) {
            commandRequest.reply(new Failure(101, "timeout"));
        } catch (InterruptedException ignored) {
            commandRequest.reply(new Failure(100, "error"));
        }
    }

    /**移除节点
     * @param commandRequest
     */
    public void removeNode(CommandRequest<RemoveNodeCommand> commandRequest) {
        Redirect redirect = checkLeadership();
        if (redirect != null) {
            commandRequest.reply(redirect);
            return;
        }

        RemoveNodeCommand command = commandRequest.getCommand();
        GroupConfigChangeTaskReference taskReference = node.removeNode(command.getNodeId());
        awaitResult(taskReference, commandRequest);
    }

    public void set(CommandRequest<SetCommand> commandRequest) {
    	// 如果当前节点不是leader节点，则返回Redirect
        Redirect redirect = checkLeadership();
        if (redirect != null) {
            commandRequest.reply(redirect);
            return;
        }

        SetCommand command = commandRequest.getCommand();
        logger.info("set {}", command.getKey());
        // 记录请求id和commandRequest的映射
        this.pendingCommands.put(command.getRequestId(), commandRequest);
        // 客户端连接关闭时从映射中移除
        commandRequest.addCloseListener(() -> pendingCommands.remove(command.getRequestId()));
        // 追加日志
        this.node.appendLog(command.toBytes());
    }

    /**在raft服务中，即使是只读请求，也要经历日志复制的过程，保证节点间的一致性后才能回复结果。可以使用之前介绍的空操作日志来实现这一要求。本书为了快速测试，跳过了日志复制，直接返回结果
     * <br/>注意，如果是生产环境，绝对不能直接跳过日志复制
     * <br/>注意，本例不再使用日志复制的方式，而是采用readIndex的方式处理，见《11优化》-《1readIndex》
     * @param commandRequest
     */
    public void get(CommandRequest<GetCommand> commandRequest) {
    	// 如果当前节点不是leader节点，则返回Redirect
        Redirect redirect = checkLeadership();
        if (redirect != null) {
        	logger.info("reply {}", redirect);
            commandRequest.reply(redirect);
            return;
        }

        GetCommand command = commandRequest.getCommand();
        logger.info("get {}", command.getKey());
        // 放入待机命令组
        pendingCommands.put(command.getRequestId(), commandRequest);
        // 把当前请求加入ReadIndex队列
        this.node.enqueueReadIndex(command.getRequestId());
    }

    private Redirect checkLeadership() {
        RoleNameAndLeaderId state = node.getRoleNameAndLeaderId();
        if (state.getRoleName() != RoleName.LEADER) {
            return new Redirect(state.getLeaderId());
        }
        return null;
    }

    static void toSnapshot(Map<String, byte[]> map, OutputStream output) throws IOException {
    	HessianOutput hessianOutput = new HessianOutput(output);
    	hessianOutput.writeObject(map);
    }

    @SuppressWarnings("unchecked")
	static Map<String, byte[]> fromSnapshot(InputStream input) throws IOException {
    	HessianInput hessianInput = new HessianInput(input);
        return (Map<String, byte[]>)hessianInput.readObject();
    }

    private class StateMachineImpl extends AbstractSingleThreadStateMachine {

		@Override
		protected void onReadIndexReached(String requestId) {
			CommandRequest<?> commandRequest = pendingCommands.remove(requestId);
			if(commandRequest != null) {
				GetCommand command = (GetCommand)commandRequest.getCommand();
				commandRequest.reply(new GetCommandResponse(map.get(command.getKey())));
			}
		}

        @Override
        protected void applyCommand(@Nonnull byte[] commandBytes) {
        	// 恢复命令
            SetCommand command = SetCommand.fromBytes(commandBytes);
            // 修改数据
            map.put(command.getKey(), command.getValue());
            // 查找连接
            CommandRequest<?> commandRequest = pendingCommands.remove(command.getRequestId());
            if (commandRequest != null) {
            	// 回复连接
                commandRequest.reply(Success.INSTANCE);
            }
        }

        @Override
        protected void doApplySnapshot(@Nonnull InputStream input) throws IOException {
            map = fromSnapshot(input);
            // 关闭文件流(自己加的，防止installing最后一步修改文件名时失败)
            input.close();
        }

        @Override
        public boolean shouldGenerateSnapshot(int firstLogIndex, int lastApplied) {
        	// 2条之后生成日志快照
            return lastApplied - firstLogIndex > 1;
        }

        @Override
        public void generateSnapshot(@Nonnull OutputStream output) throws IOException {
            toSnapshot(map, output);
        }

    }

}
